package com.atguigu.edu.realtime.app.dwd.db;

import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdUserRegister {
    public static void main(String[] args) {
        //TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //TODO 2 检查点相关设置
        /*//2.1 开启检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        //2.2 设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        //2.3 设置job取消之后，检查点是否保留
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.4 设置两个检查点之间最小时间间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        //2.5 设置重启策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));
        //2.6 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop101:8020/edu/ck");
        //2.7 设置操作hadoop的用户
        System.setProperty("HADOOP_USER_NAME","atguigu");*/
        //TODO 3 从topic_db中读取数据，并创建动态表
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("dwd_user_register_group"));
        //TODO 4 筛选出用户表且是用户注册的数据
        Table userInfo = tableEnv.sqlQuery("select\n" +
                "data['id'] user_id,\n" +
                "data['create_time'] create_time,\n" +
                "ts\n" +
                "from topic_db\n" +
                "where `table` = 'user_info'\n" +
                "and `type` = 'insert'\n");
        tableEnv.createTemporaryView("user_info", userInfo);
        // TODO 5 创建 Upsert-Kafka dwd_user_register 表
        tableEnv.executeSql("create table `dwd_user_register`(\n" +
                "`user_id` string,\n" +
                "`date_id` string,\n" +
                "`create_time` string,\n" +
                "`ts` string,\n" +
                "primary key(`user_id`) not enforced\n" +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_user_register"));
        // TODO 6 将用户注册数据通过upsert-kafka连接器写入kafka
        tableEnv.executeSql("insert into dwd_user_register\n" +
                "            select \n" +
                "            user_id,\n" +
                "            date_format(create_time, 'yyyy-MM-dd') date_id,\n" +
                "            create_time,\n" +
                "            ts\n" +
                "            from user_info");
        tableEnv.executeSql("select * from dwd_user_register").print();
    }
}
